package com.hilin.hilinim.netty.ws;

import com.hilin.common.util.ThreadPoolUtils;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import mousio.etcd4j.EtcdClient;
import mousio.etcd4j.responses.EtcdKeysResponse;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.net.URI;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class WsConnector {

    private static final ThreadPoolExecutor POOL_EXECUTOR = ThreadPoolUtils.createThreadPool("im_thread_", 1, 1, 3600, 100);

    private static final String HTTP_LINE = "GET /v2/keys/?stream=true&wait=true&recursive=true HTTP/1.1\n" +
            "host: 127.0.0.1\n" +
            "connection: keep-alive\n\n";

    @Autowired
    private WsConfig wsConfig;

    @Autowired
    private WsHandler wsHandler;

    /**
     * 启动ws服务
     */
    public void start() {
        POOL_EXECUTOR.execute(
                () -> {
                    try {
                        EventLoopGroup bossGroup = new NioEventLoopGroup();
                        EventLoopGroup group = new NioEventLoopGroup();
                        try {
                            ServerBootstrap sb = new ServerBootstrap();
                            sb.option(ChannelOption.SO_BACKLOG, 1024);
                            sb.option(ChannelOption.SO_REUSEADDR, true);
                            sb.group(group, bossGroup)
                                    // 绑定线程池
                                    .channel(NioServerSocketChannel.class)
                                    .localAddress("0.0.0.0", wsConfig.getPort())
                                    .childHandler(new ChannelInitializer<SocketChannel>() {
                                        @Override
                                        protected void initChannel(SocketChannel ch) throws Exception {
                                            //websocket协议本身是基于http协议的，所以这边也要使用http解编码器
                                            ch.pipeline().addLast(new HttpServerCodec());
                                            //以块的方式来写的处理器
                                            ch.pipeline().addLast(new ChunkedWriteHandler());
                                            // 如果未启动文件服务，则设置http请求数据大小为8192
                                            ch.pipeline().addLast(new HttpObjectAggregator(81920));
                                            ch.pipeline().addLast(new IdleStateHandler(
                                                    wsConfig.getHeartbeatTimeout(),
                                                    wsConfig.getHeartbeatTimeout(),
                                                    wsConfig.getHeartbeatTimeout(),
                                                    TimeUnit.SECONDS));
                                            ch.pipeline().addLast(wsHandler);
                                            ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", null, true, 65536 * 10));
                                        }
                                    });
                            // 服务器异步创建绑定
                            ChannelFuture cf = sb.bind().sync();
                            log.info("[WS连接]启动正在监听： " + cf.channel().localAddress());
                            // 关闭服务器通道
                            cf.channel().closeFuture().sync();
                        } finally {
                            // 释放线程池资源
                            group.shutdownGracefully().sync();
                            bossGroup.shutdownGracefully().sync();
                        }
                    } catch (Exception e) {
                        log.error("[WS连接]启动im失败", e);
                    }
                });
    }

    public static void main(String[] args) throws Exception {
        EtcdClient etcdClient = new EtcdClient(new URI("http://127.0.0.1:2379"));
        long waitIndex = 99999;
        while (true) {
            EtcdKeysResponse response = null;
            if (waitIndex <= 0) {
                response = etcdClient.get("/test/a")
                        .timeout(-1, TimeUnit.SECONDS)
                        .send().get();
            } else {
                response = etcdClient.get("/test/a")
                        .waitForChange(waitIndex)
                        .timeout(-1, TimeUnit.SECONDS)
                        .send().get();
            }
            waitIndex = response.node.modifiedIndex + 1;
            System.out.println(response.node.value);
        }

//        try {
//            EventLoopGroup group = new NioEventLoopGroup();
//            Bootstrap bootstrap = new Bootstrap();
//            bootstrap.group(group);
//            bootstrap.channel(NioSocketChannel.class);
//            bootstrap.remoteAddress("127.0.0.1", 2379);
//            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
//                @Override
//                protected void initChannel(SocketChannel ch) throws Exception {
//                    ch.pipeline().addLast(new SimpleChannelInboundHandler<ByteBuf>() {
//                        @Override
//                        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
//                            System.out.println(convertByteBufToString(msg));
//                        }
//                    });
//                }
//            });
//            ChannelFuture future = bootstrap.connect().sync();
//            future.channel().writeAndFlush(Unpooled.copiedBuffer(HTTP_LINE.getBytes(StandardCharsets.UTF_8))).sync();
//        } catch (Exception e) {
//            log.error("[WS连接]启动im失败", e);
//        }
    }

    public static String convertByteBufToString(ByteBuf buf) {
        String str;
        if (buf.hasArray()) { // 处理堆缓冲区
            str = new String(buf.array(), buf.arrayOffset() + buf.readerIndex(), buf.readableBytes());
        } else { // 处理直接缓冲区以及复合缓冲区
            byte[] bytes = new byte[buf.readableBytes()];
            buf.getBytes(buf.readerIndex(), bytes);
            str = new String(bytes, 0, buf.readableBytes());
        }
        return str;
    }

}
